package com.itsu.job.rpc.basic;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.json.JSONUtil;
import com.itsu.job.components.Const;
import com.itsu.job.exception.JobException;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.*;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author Jerry.Su
 * @Date 2022/1/24 11:09
 */
public abstract class EmbedHttpServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {

    private static final Logger logger = LoggerFactory.getLogger(EmbedHttpServerHandler.class);

    public static final List<HttpMethod> allowedMethods =
            CollUtil.newArrayList(HttpMethod.DELETE, HttpMethod.GET, HttpMethod.POST, HttpMethod.PUT);

    private final String accessToken;
    private final ThreadPoolExecutor bizThreadPool;
    private String responseContentType = "application/json;charset=UTF-8";

    public EmbedHttpServerHandler(String accessToken, ThreadPoolExecutor bizThreadPool) {
        this.accessToken = accessToken;
        this.bizThreadPool = bizThreadPool;
    }

    public void setResponseContentType(String responseContentType) {
        this.responseContentType = responseContentType;
    }

    @Override
    protected void channelRead0(final ChannelHandlerContext ctx, FullHttpRequest msg) throws Exception {

        // request parse
        //final byte[] requestBytes = ByteBufUtil.getBytes(msg.content());    // byteBuf.toString(io.netty.util.CharsetUtil.UTF_8);
        String requestData = msg.content().toString(CharsetUtil.UTF_8);
        String uri = msg.uri();
        HttpMethod httpMethod = msg.method();
        boolean keepAlive = HttpUtil.isKeepAlive(msg);
        String accessTokenReq = getToken(msg);

        // invoke
        bizThreadPool.execute(() -> {
            // do invoke
            Object responseObj = null;
            String responseJson = null;
            try {
                responseObj = process(httpMethod, uri, requestData, accessTokenReq);
                responseJson = JSONUtil.toJsonStr(responseObj);
                writeResponse(ctx, keepAlive, responseJson);
            } catch (Exception e) {
                if (e instanceof JobException) {
                    logger.info("netty rpc calling failed cause by: {}", e.getMessage());
                } else {
                    logger.error("unexpected error occur", e);
                }
                Map<Object, Object> response = MapUtil.builder().put("code", -1)
                        .put("msg", e.getMessage()).build();
                responseJson = JSONUtil.toJsonStr(response);
                writeResponse(ctx, keepAlive, responseJson, HttpResponseStatus.INTERNAL_SERVER_ERROR);
            }
            logger.debug("response: [{}]", responseJson);
        });
    }

    protected String getToken(FullHttpRequest msg) {
        return msg.headers().get(Const.TOKEN_HEADER_KEY);
    }

    protected Object process(HttpMethod httpMethod, String uri, String requestData, String accessTokenReq) {
        logger.debug("uri:{}, with method:{} start to process", uri, httpMethod);
        // valid
        if (!allowedMethods.contains(httpMethod)) {
            return "invalid request method";
        }
        if (uri == null || uri.trim().length() == 0) {
            return "invalid request, uri-mapping empty.";
        }
        if (accessToken != null
                && accessToken.trim().length() > 0
                && !accessToken.equals(accessTokenReq)) {
            return "The access token is wrong.";
        }

        Object result = null;
        if (HttpMethod.GET == httpMethod) {
            result = doGet(uri);
        } else if (HttpMethod.POST == httpMethod) {
            result = doPost(uri, requestData);
        } else if (HttpMethod.PUT == httpMethod) {
            result = doPut(uri, requestData);
        } else if (HttpMethod.DELETE == httpMethod) {
            result = doDelete(uri, requestData);
        } else {
            throw new RuntimeException("not supported http method: " + httpMethod);
        }
        return result;
    }

    protected abstract Object doDelete(String uri, String requestData);

    protected abstract Object doPut(String uri, String requestData);

    protected abstract Object doPost(String uri, String requestData);

    protected abstract Object doGet(String uri);

    /**
     * write response
     */
    protected void writeResponse(ChannelHandlerContext ctx, boolean keepAlive, String responseJson, HttpResponseStatus status) {
        FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, status,
                Unpooled.copiedBuffer(responseJson, CharsetUtil.UTF_8));   //  Unpooled.wrappedBuffer(responseJson)
        response.headers().set(HttpHeaderNames.CONTENT_TYPE, responseContentType);       // HttpHeaderValues.TEXT_PLAIN.toString()
        response.headers().set(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
        if (keepAlive) {
            response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
        }
        ctx.writeAndFlush(response);
    }

    protected void writeResponse(ChannelHandlerContext ctx, boolean keepAlive, String responseJson) {
        writeResponse(ctx, keepAlive, responseJson, HttpResponseStatus.OK);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

//    @Override
//    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
//        logger.error(">>>>>>>>>>>  netty_http server caught exception", cause);
//        handleException(cause);
//        ctx.close();
//    }

    protected abstract void handleException(Throwable cause);

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            ctx.channel().close();      // beat 3N, close if idle
            logger.debug(">>>>>>>>>>>  netty_http server close an idle channel.");
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}